import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;



import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


import javax.security.auth.login.AppConfigurationEntry;

import java.io.File;
import java.io.IOException;

/**
 * 注意整体的包，很容易引错
 * 此程序分2部分，map部分和redeuce部分
 */
public class WordCount {
    /**
     * Map部分
     *
     * 例如有一个文件，名为 测试 其内容为a,b,c
     *                                   a,e,d
     * 下面的LongWritable是偏移量，意思是从哪里开始读，后面的Text是第一行的内容，他是一行一行的读。
     * 在后面的Text以及IntWritable是Reduce要发出去的东西。
     *
     */
    public static class MapWordCount extends Mapper<LongWritable, Text,Text, IntWritable>{

        @Override
        /**
         * 重写方法，写上prote自动会提示
         */
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            /**
             * 上面的key 是a,value是1.
             * 要把value转为String类型
             */
            String s = value.toString();
            /**
             * 把文件里的内容进行切分，以逗号分开
             */
            String[] line = s.split(",");
            /**
             * 用for循环来读取内容，一行行读，下面的s1是第一行的内容，IntWritable是值，如本文件的a是1，b是1，c是1.
             */
            for (String s1 : line){
                context.write(new Text(s1),new IntWritable(1));
            }
        }
    }

    /**
     * Reduce部分
     */
    public static class ReduceWordCount extends Reducer<Text,IntWritable,Text,IntWritable> {
        /**
         *
         * @param key  Text 第二个Text是最后生成的内容
         * @param values   IntWritable  第二个IntWritable是 最后的值例如a出现了2次，这个a值为2
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int a = 0;
            /**
             * 需要循环来读取，例如a一共出现了2次用下面的for循环遍历的方式
             */
            for (IntWritable value : values){
                a = a + value.get();
            }
            context.write(key,new IntWritable(a));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        /**
         * 要连接本地文件
         */
        Configuration configuration = new Configuration();
        Job job =  Job.getInstance(configuration);

        job.setJarByClass(WordCount.class);
        job.setMapperClass(MapWordCount.class);
        job.setReducerClass(ReduceWordCount.class);
/**
 * 读取的内容和值
 */
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
/**
 * 最后输出的内容和值例如 a  2
 */
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
/**
 * 判断文件是否存在
 */
        File file = new File("E:/测试/output");
        if (file.exists()){
            FileUtil.fullyDelete(file);
        }
        /**
         * 文件的初始路径以及最后的输出路径
         */
        FileInputFormat.addInputPath(job,new Path("E:/测试/input"));
        FileOutputFormat.setOutputPath(job,new Path("E:/测试/output"));

         job.setNumReduceTasks(1);
         System.exit(job.waitForCompletion(true)?0:1);


    }



}
